package com.gugee.common.task.service;

import com.gugee.common.task.dto.TaskParam;
import com.gugee.common.task.dal.service.TaskRecordSvc;
import com.gugee.common.task.model.TaskRecord;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Optional;

/**
 * Created with IntelliJ IDEA.
 * Description:查询需要的记录，构造消息并发送至主题中
 * User: luolifeng
 * Date: 2020-10-15 17:12
 */
@Slf4j
public abstract class AbstractProducerHandler<T> {
    @Autowired
    public TaskRecordSvc taskRecordSvc;


    public ReturnT<String> xxlJobStart(Integer taskType, String who, TaskParam taskParam, boolean topN){
        long start = System.currentTimeMillis();
        try {
            log.info("{}:开始启动消息投递任务: taskParam={}",who, taskParam);
            XxlJobLogger.log("{}:开始启动消息投递任务: taskParam={}", who,taskParam);
            if (taskType!=null &&  !topN) {
                sendMsg2Kafka(taskType, who, taskParam);
            }

            return ReturnT.SUCCESS;
        } catch (Exception ex) {
            XxlJobLogger.log(ex);
            ex.printStackTrace();
            return ReturnT.FAIL;
        }finally {
            log.info("{}:投递结束  end cost total={},",who, (System.currentTimeMillis() - start) / 1000);
            XxlJobLogger.log("{}:投递结束  end cost total={}, params={}", who,(System.currentTimeMillis() - start) / 1000,taskParam);
        }
    }
    /**
     * 获取当前需采集的页数
     * @param taskType 字典类型值，对应某种业务数据类别
     * @param who 操作方
     * @return
     */
   public TaskRecord getTaskRecord(int taskType, String who) {

        Optional<TaskRecord> taskRecordOptional = taskRecordSvc.findByTaskType(taskType);
       TaskRecord taskRecord;
        if (taskRecordOptional.isPresent()) {
            taskRecord = taskRecordOptional.get();
        } else {
            taskRecord = new TaskRecord();
            taskRecord.setTaskType(taskType);
            taskRecord.setUpdateBy(who);
            taskRecord.setPreStartObjectId(0L);
            taskRecord.setPreEndObjectId(0L);
            taskRecord.setCurrentPage(1);
            taskRecord.setLastRecords(0L);
            taskRecord.setRecords(0L);
            taskRecord.setLastTotalPages(0);
            taskRecord.setFirstPageStartTimeMills(0L);
            taskRecordSvc.save(taskRecord);
        }
        return taskRecord;
    }

    /**
     * 查询具体页内的记录
     * @param taskRecord 业务类型数据的采集页
     * @param taskParam  任务参数，包含了每页的页码大小
     * @return
     */
    public abstract List<T> findMessageObject(TaskRecord taskRecord,TaskParam taskParam);

    /**
     * 返回目标对象在库表中的总记录数 为跳页做准备
     */
    public abstract long objectTotals();

    public abstract void sendMsg(List<T> messages,String topic);

    /**
     * 发送
     * @param taskType 任务类型
     * @param who 调用者
     * @param taskParam 任务参数
     */
    public void sendMsg2Kafka(int taskType, String who,TaskParam taskParam){
        sendMsg2Kafka(taskType,who,taskParam,20);
    }
    public boolean isJump(TaskRecord taskRecord){
        return false;
    }
    /**
     * 发送
     * @param taskType 任务类型
     * @param who 调用者
     * @param taskParam 任务参数
     */
    public void sendMsg2Kafka(int taskType, String who,TaskParam taskParam,int passEmptyPageCount){
        long start=System.currentTimeMillis();
        //获取将发送的位置
        TaskRecord taskRecord =getTaskRecord(taskType,who);
        if(isJump(taskRecord)){
            return;
        }
        if(taskParam.getStartPage()==0) {
            taskParam.setStartPage(1);
        }
        String currentTime= LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        if(taskRecord.getCurrentPage()==1){
            taskRecord.setFirstPageStartTime(currentTime);
            taskRecord.setFirstPageStartTimeMills(System.currentTimeMillis());
            taskRecord.setRecords(0L);
        }
        long totals=objectTotals();
        List<T> messages =null;
        //最多尝试跳过10次空页
        for(int i = 0;i < passEmptyPageCount; i++) {
            //总记录数比较，获取到了末尾说明本轮结束
            if(totals<(taskRecord.getCurrentPage()-1)*taskParam.getRecords() || taskRecord.getCurrentPage()>taskParam.getPagesLimit()){
                taskRecord.setLastTotalPages(taskRecord.getCurrentPage());
                taskRecord.setCurrentPage(0);
                break;
            }
            //获取应该发送的消息
              messages = findMessageObject(taskRecord, taskParam);
              if(messages==null|| !messages.isEmpty()){
                  break;
              }


            taskRecord.setCurrentPage(taskRecord.getCurrentPage()+1);
        }
            if(messages==null){
                messages =new ArrayList<>(0);
            }
        //记录此次的发送情况
        log.info(" who={}, size={},page={}, pageSize={},  objectId from  {} to {},times={}秒",who,messages.size(),taskRecord.getCurrentPage(),taskParam.getRecords(),taskRecord.getPreStartObjectId(),taskRecord.getPreEndObjectId(),(System.currentTimeMillis()-start)/1000);
        XxlJobLogger.log("  size={},page={}, pageSize={},  record from  {} to {},times={}秒",messages.size(),taskRecord.getCurrentPage(),taskParam.getRecords(),taskRecord.getPreStartObjectId(),taskRecord.getPreEndObjectId(),(System.currentTimeMillis()-start)/1000);



        //记录下一次发送的位置
        int nextPage=taskRecord.getCurrentPage()%taskParam.getPagesLimit()+1;

        sendMsg(messages,taskParam.getTopic());
        taskRecord.setRecords(taskRecord.getRecords()+messages.size());
        //如果下一页是重新开始，则设置本轮的耗时
        if(nextPage==1){

            taskRecord.setLastTotalSpendTime((int)((System.currentTimeMillis()-taskRecord.getFirstPageStartTimeMills())/1000/60));
            taskRecord.setLastRecords(taskRecord.getRecords());
            taskRecord.setRecords(0L);
        }
        //发送消息后，设置下一次待发送页数
        taskRecord.setCurrentPage(nextPage);
        taskRecord.setUpdateTime(new Date());
        taskRecordSvc.save(taskRecord);

    }

}
